package com.nokia.ion.sdu.cdr_decoder;

import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.node.Node;
import org.elasticsearch.threadpool.ThreadPool;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by vicwang on 2017/9/19.
 */
public class ElasticsearchWriter extends CdrWriter {

    private RestClient lowLevelRestClient;
    private RestHighLevelClient client;

    private Configuration.ElasticsearchConfigure elasticsearchConfigure;

    private BulkProcessor processor;

    public ElasticsearchWriter(LinkedBlockingQueue<Cdr> queue, Configuration configuration) throws IOException {
        super(queue, configuration);

        this.elasticsearchConfigure = configuration.elasticsearch;

        lowLevelRestClient = RestClient.builder(new HttpHost(elasticsearchConfigure.url)).build();
        client = new RestHighLevelClient(lowLevelRestClient);

        ThreadPool threadPool = new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "high-level-client").build());
        BulkProcessor.Builder builder = new BulkProcessor.Builder(client::bulkAsync, listener, threadPool);
//        builder.setBulkActions(4000);
        builder.setBulkSize(new ByteSizeValue((long) elasticsearchConfigure.bulkSize, ByteSizeUnit.MB));
        builder.setConcurrentRequests(elasticsearchConfigure.concurrentRequests);
//        builder.setFlushInterval(TimeValue.timeValueSeconds(10L));
//        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));

        processor = builder.build();
    }

    private static final BulkProcessor.Listener listener = new BulkProcessor.Listener() {

        @Override
        public void beforeBulk(long l, BulkRequest bulkRequest) {

        }

        @Override
        public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
            if (bulkResponse.hasFailures()) {
                String msg = bulkResponse.buildFailureMessage();
                System.err.println(String.format("Bulk(%d) executed with failures. %s", l, msg));
            }
        }

        @Override
        public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
            System.err.println(throwable);
            System.err.println();
        }
    };

    public void write(Cdr cdr) {
        processor.add(getRequest(cdr));
    }

    void close() {
        try {
            processor.close();
            lowLevelRestClient.close();
        } catch (IOException e) {
            System.err.println(e);
        }
    }

    private IndexRequest getRequest(Cdr cdr) {
        Map<String, Object> jsonMap = getMap(cdr);

        return new IndexRequest("cas-55638-c6b7", "cdr_test").source(jsonMap);
    }

    private Map<String, Object> getMap(Cdr cdr) {
        Map<String, Object> jsonMap = new HashMap<>();
        jsonMap.put("node-id", cdr.nodeID);
        jsonMap.put("s-addr", cdr.servingNodeAddress);
        jsonMap.put("imsi", cdr.servedIMSI);
        jsonMap.put("rat-type", cdr.rATType);
        jsonMap.put("uli", cdr.userLocationInformation);
        jsonMap.put("open_time", Cdr.localFormat.format(cdr.recordOpeningTime));
        jsonMap.put("duration", cdr.duration);
        jsonMap.put("cause", cdr.causeForRecClosing);
        jsonMap.put("charging-id", cdr.chargingId);

        return jsonMap;
    }
}
